package com.shujia.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

public class IndexHBaseCoprocessor extends BaseRegionObserver {


    /**
     * postPut  在put之后执行
     * 在这里增加索引
     */
    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {

        try {
            //往索引表中增加一个索引
            //创建hadoop配置文件对象
            Configuration conf = new Configuration();

            //指定zk连接地址
            conf.set("hbase.zookeeper.quorum", "node2:2181,node3:2181,node1:2181");
            //增删改数据使用
            HConnection connection = HConnectionManager.createConnection(conf);

            HTableInterface student_index = connection.getTable("student_index");

            //取出班级
            List<Cell> cells = put.get("info".getBytes(), "clazz".getBytes());
            Cell cell = cells.get(0);

            byte[] bytes = CellUtil.cloneValue(cell);
            String clazz = Bytes.toString(bytes);


            String id = Bytes.toString(CellUtil.cloneRow(cell));


            Put indexPut = new Put(clazz.getBytes());
            indexPut.add("i".getBytes(), id.getBytes(), "".getBytes());

            //插入索引
            student_index.put(indexPut);


            connection.close();
        } catch (Exception e1) {
            e1.printStackTrace();
        }


    }

}
